package com.talent.core.protocol.mqtt.store;

import com.talent.core.protocol.mqtt.TlMqttSession;
import com.talent.core.protocol.mqtt.model.entity.TlSubClient;
import com.talent.core.protocol.mqtt.model.entity.TlMessage;
import com.talent.core.protocol.mqtt.model.request.TlMqttPubRecReq;
import com.talent.core.protocol.mqtt.model.request.TlMqttPublishReq;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName：TlMqttLocalStore
 * @Author: hszhou
 * @Date: 2024/12/25 13:34
 * @Description: mqtt的相关数据存储在本地
 */
public class TlMqttLocalStore implements IMqttStore {

    private static final Logger logger = LoggerFactory.getLogger(TlMqttLocalStore.class);
    /**
     * 对应的是订阅的主题  后面是订阅主题的客户端集合
     */
    public static final ConcurrentHashMap<String, Set<TlSubClient>> TOPIC_TO_CLIENT = new ConcurrentHashMap<>();

    /**
     * 对应的是客户端的ID后面是客户端对应的session
     */
    public static final ConcurrentHashMap<String, TlMqttSession> CLIENT_TO_SESSION=new ConcurrentHashMap<String, TlMqttSession>();


    /**
     * 对应的主题与保留消息
     */
    public static final ConcurrentHashMap<String, TlMessage> WILL_MESSAGE = new ConcurrentHashMap<>();

    /**
     * broker用于存储本地的消息
     */
    public static final ConcurrentHashMap<Integer,TlMqttPublishReq> BROKER_MESSAGE=new ConcurrentHashMap<>();


    public static final ConcurrentHashMap<Integer, TlMqttPubRecReq> PUBREC_MAP=new ConcurrentHashMap<>();

    /**
     * 用于生成消息ID
     */
    private final static AtomicInteger IDENTIFIER_MESSAGE = new AtomicInteger(0);

    @Override
    public void putSubClient(TlSubClient client) {
        String topic = client.getTopic();
        Set<TlSubClient> tlSubClients = TOPIC_TO_CLIENT.get(topic);
        if(tlSubClients==null){
            tlSubClients = new HashSet<>();
        }
        tlSubClients.add(client);
        TOPIC_TO_CLIENT.put(topic,tlSubClients);
    }

    @Override
    public List<TlSubClient> getSubClientByTopic(String topic) {
        List<TlSubClient> rs = new ArrayList<>();
        TOPIC_TO_CLIENT.forEach((k,v)->{
            if (match(topic,k)) {
                rs.addAll(v);
            }
        });
        return rs;
    }

    @Override
    public List<TlSubClient> getSubClientByTopicAndQos(String topic, int qos) {
        List<TlSubClient> rs = new ArrayList<>();
        TOPIC_TO_CLIENT.forEach((k,v)->{
            if (match(topic,k)) {
                rs.addAll(v.stream().filter(e->e.getQos()==qos)
                        .toList());
            }
        });
        return rs;
    }

    @Override
    public void removeSubClientByTopic(String topic,String clientId) {


        Set<TlSubClient> tlSubClients = TOPIC_TO_CLIENT.get(topic);
        if(null==tlSubClients){
            return;
        }

        Iterator<TlSubClient> iterator = tlSubClients.iterator();
        while (iterator.hasNext()){
            TlSubClient next = iterator.next();
            if(next.getClientId().equals(clientId)){
                iterator.remove();
            }
        }
        tlSubClients.remove(topic);

    }

    @Override
    public void putSession(String clientId, TlMqttSession session) {
        CLIENT_TO_SESSION.put(clientId,session);
    }

    @Override
    public TlMqttSession getSessionByClientId(String clientId) {
        return CLIENT_TO_SESSION.get(clientId);
    }

    @Override
    public TlMqttSession removeSessionByClientId(String clientId) {
       return CLIENT_TO_SESSION.remove(clientId);
    }

    @Override
    public TlMqttPublishReq getRetainMessageByTopic(String topic) {

        /**
         * sebor/1 message1
         * sebor/2 message2
         * 订阅sebor/+  会受到message2
         * 订阅sebor/# 会受到message2
         */
        TlMessage retainMessage=null;
        Set<Map.Entry<String, TlMessage>> entries = WILL_MESSAGE.entrySet();
        for (Map.Entry<String, TlMessage> entry : entries) {
            if(match(entry.getKey(),topic)){
                if(retainMessage==null){
                    retainMessage=entry.getValue();
                }else if(entry.getValue().compareTo(retainMessage)>0){
                    retainMessage = entry.getValue();
                }
            }
        }
        if(retainMessage==null){
            return null;
        }
        return retainMessage.getReq();
    }

    @Override
    public TlMessage putRetainMessage(String topic, TlMessage retainMessage) {
        return WILL_MESSAGE.put(topic,retainMessage);
    }

    @Override
    public void brokerStoreMsg(int messageId,TlMqttPublishReq req ) {

        BROKER_MESSAGE.put(messageId,req);
    }

    @Override
    public void removeStoreMsg(int messageId) {
        BROKER_MESSAGE.remove(messageId);
    }

    @Override
    public TlMqttPublishReq getStoreMsg(int messageId) {
        return BROKER_MESSAGE.get(messageId);
    }

    @Override
    public void storePubRec(int messageId, TlMqttPubRecReq req) {

        PUBREC_MAP.put(messageId,req);
    }

    @Override
    public void removePubRec(int messageId) {
        PUBREC_MAP.remove(messageId);
    }

    @Override
    public TlMqttPubRecReq getPubRec(int messageId) {
       return PUBREC_MAP.get(messageId);
    }

    @Override
    public Integer getMessageId() {
      return IDENTIFIER_MESSAGE.getAndIncrement();
    }


}
